package com.zt.bigdata.elasticsearch.rest;

import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.*;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortOrder;

import java.util.Arrays;

import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
import static org.elasticsearch.index.query.QueryBuilders.matchQuery;

/**
 * @ClassName MoveData
 * @Description
 * @Author zhangtonghy
 * @Date 2019-04-08 11:18
 * @Copyright: 版权所有 (C) zt zt.
 * @注意：本内容仅限于zt内部传阅，禁止外泄以及用于其他的商业目的
 **/
public class MoveData {
    private static final String SOURCE_HOSTNAME = "";
    private static final String SOURCE_USERNAME = "";
    private static final String SOURCE_PASSWORD = "";
    private static final int SOURCE_PORT = 6386;

    private static final String TARGET_HOSTNAME = "";
    private static final String TARGET_USERNAME = "";
    private static final String TARGET_PASSWORD = "";
    private static final int TARGET_PORT = 9200;


    public RestHighLevelClient getClient(String username, String password, String hostname, int port) {
        RestHighLevelClient restHighLevelClient = null;
        if (null == restHighLevelClient) {
            System.out.println("创建客户端");

            final CredentialsProvider credentialsProvider =
                    new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY,
                    new UsernamePasswordCredentials(username, password));


            RestClientBuilder builder = RestClient.builder(
                    new HttpHost(hostname, port))

                    .setRequestConfigCallback(
                            new RestClientBuilder.RequestConfigCallback() {
                                @Override
                                public RequestConfig.Builder customizeRequestConfig(
                                        RequestConfig.Builder requestConfigBuilder) {
                                    return requestConfigBuilder
                                            .setConnectTimeout(5000)
                                            .setSocketTimeout(60000);
                                }
                            })
                    .setMaxRetryTimeoutMillis(60000)
                    .setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder
                            .setDefaultCredentialsProvider(credentialsProvider));


            restHighLevelClient = new RestHighLevelClient(builder);
            return restHighLevelClient;
        }
        return restHighLevelClient;
    }

    public static void main(String[] args) throws Exception {
        String index = "cdn-file-dispatch-2019.04.01";
        MoveData moveData = new MoveData();
        RestHighLevelClient source = moveData.getClient(SOURCE_USERNAME, SOURCE_PASSWORD, SOURCE_HOSTNAME, SOURCE_PORT);
        RestHighLevelClient target = moveData.getClient(TARGET_USERNAME, TARGET_PASSWORD, TARGET_HOSTNAME, TARGET_PORT);


        final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(10L));
        SearchRequest searchRequest = new SearchRequest(index);
        searchRequest.scroll(scroll);

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(matchAllQuery());
        searchRequest.source(searchSourceBuilder);

        SearchResponse searchResponse = source.search(searchRequest, RequestOptions.DEFAULT);
        String scrollId = searchResponse.getScrollId();
        SearchHit[] searchHits = searchResponse.getHits().getHits();


        moveData.asyncBulkInsert(searchHits, target);

        while (searchHits != null && searchHits.length > 0) {
            SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
            scrollRequest.scroll(scroll);
            searchResponse = source.scroll(scrollRequest, RequestOptions.DEFAULT);
            scrollId = searchResponse.getScrollId();
            searchHits = searchResponse.getHits().getHits();

            moveData.asyncBulkInsert(searchHits, target);
        }

        ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
        clearScrollRequest.addScrollId(scrollId);
        ClearScrollResponse clearScrollResponse = source.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        boolean succeeded = clearScrollResponse.isSucceeded();


    }

    public void asyncBulkInsert(SearchHit[] searchHits, RestHighLevelClient client) {
        BulkRequest request = new BulkRequest();
        for (int i = 0; i < searchHits.length; i++) {
            request.add(new IndexRequest(searchHits[i].getIndex(), searchHits[i].getType())
                    .source(searchHits[i].getSourceAsMap()));
        }
        client.bulkAsync(request, RequestOptions.DEFAULT, new ActionListener<BulkResponse>() {
            @Override
            public void onResponse(BulkResponse bulkItemResponses) {

            }

            @Override
            public void onFailure(Exception e) {
                System.out.println("\n\n" + e);
            }
        });
    }
}
